"""
线程池的使用
- 适合 I/O 密集型或高并发的小任务（注意 GIL，CPU 密集型优先使用进程池）。
示例包含：
1) map 示例（按输入顺序返回结果）
2) submit + 回调（异步提交单个任务并处理完成/异常）
3) as_completed（流式处理已完成任务）
4) 结果超时示例
5) 共享可变状态示例（用 Lock 保护）
"""
from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError
import threading
import time
import random

def worker(x, delay_range=(0.1, 0.6)):
    """模拟工作：随机睡眠后返回 x*x"""
    print(f"线程名称:{threading.current_thread().name} 处理输入: x= {x}, delay_range={delay_range}")
    sleep_time = random.uniform(*delay_range)
    print(f"线程名称:{threading.current_thread().name} 睡眠 {sleep_time:.2f} 秒")
    time.sleep(sleep_time)  # 模拟 I/O 操作或耗时计算
    return x * x

def simple_map_example():
    """使用 ThreadPoolExecutor.map 并行计算，返回顺序与输入一致。"""
    inputs = list(range(8)) # 结果 [0, 1, 2, 3, 4, 5, 6, 7]
    delay_ranges = [(1, 5) for _ in inputs]
    delay_ranges[3] = (5, 10)  # 故意让第4个任务延迟更长一些
    # 将 inputs 和 delay_ranges 打包成元组列表传递给 worker, 线程数量为4个
    with ThreadPoolExecutor(max_workers=4) as exe:
        # map 会阻塞直到所有任务完成，返回按输入顺序的结果迭代器
        results = list(exe.map(worker, inputs, delay_ranges))
    print("map results:", results)

def submit_with_callback_example():
    """使用 submit 提交任务并使用 done callback 处理完成或异常。
    
    说明：
    - exe.submit(...) 返回一个 concurrent.futures.Future 对象，代表异步执行的结果。
    - fut.add_done_callback(cb) 会在 Future 完成（成功或异常）时调用回调函数 cb。
      回调函数是在设置结果/异常的线程中调用（通常是执行任务的工作线程），
      因此回调应尽量简短且不阻塞，避免影响线程池中的其它任务调度。
    - 在回调内部调用 fut.result() 会返回结果或重新抛出任务执行时的异常，
      因此常用 try/except 来区分成功/失败的处理逻辑。
    """
    def done_callback(fut):
        # 回调会在 Future 完成时被执行。使用 fut.result() 获取结果或让异常重现以便处理。
        # 注意：此回调在完成任务的线程里运行（不是主线程），应避免执行耗时操作。
        try:
            res = fut.result()  # 若任务引发异常，这里会抛出相应异常
            print("任务完成，结果:", res)
        except Exception as e:
            # 可以记录异常类型和信息，或执行补偿/重试逻辑
            print("任务异常：", type(e).__name__, e)

    # 创建线程池，max_workers 表示并发执行任务的线程数（根据任务性质调整）
    with ThreadPoolExecutor(max_workers=3) as exe:
        futures = []
        # 提交多个任务并为每个 Future 注册回调
        for i in range(6):
            fut = exe.submit(worker, i)  # 提交任务，立即返回 Future
            fut.add_done_callback(done_callback)  # 注册回调，任务完成后会被调用
            futures.append(fut)

        # 等待所有 futures 完成：
        # - fut.result() 会阻塞直到该 Future 完成，若任务失败会在此重新抛出异常。
        # - 这里以简单的同步等待为例；也可以使用 as_completed / wait 等方式进行更灵活的等待/处理。
        for fut in futures:
            try:
                res = fut.result()  # 仅用于等待任务完成；已在回调中处理结果/异常
                print("主线程确认任务完成，结果:", res)
            except Exception:
                # 已在回调中处理异常，这里可以根据需要做额外处理或忽略
                pass

def as_completed_example():
    """使用 as_completed 流式处理已完成的任务（不保证顺序）。

    说明：
    - as_completed(futures) 返回一个迭代器，按任务完成的先后顺序产生 Future。
      迭代会阻塞直到下一个 Future 完成，因此这是处理"已完成"任务的常用模式，
      非常适合需要尽快处理早完成任务的场景（例如先返回的结果优先消费）。
    - 注意：as_completed 不会改变原始 futures 列表，它只是按完成顺序产生 Future 对象。
    - 对于调试/日志通常需要知道某个 Future 对应的输入值，此处通过字典映射保存关联。
    """
    inputs = list(range(10))

    # 创建线程池并提交任务，收集 Future 列表
    with ThreadPoolExecutor(max_workers=4) as exe:
        futures = [exe.submit(worker, i) for i in inputs]

        # 可选：建立 Future -> 输入 的映射，便于在处理结果时知道是哪一个输入产生的结果
        fut_to_input = {fut: inp for fut, inp in zip(futures, inputs)}

        # as_completed 会在每个 Future 完成时立刻产出它（成功或异常）
        for fut in as_completed(futures):
            # 这里的 fut 已经完成（finished），调用 fut.result() 不会阻塞，
            # 但如果任务执行时抛出异常，result() 会重新抛出该异常，需要捕获处理。
            try:
                res = fut.result()  # 获取结果，若任务失败此处会抛出异常
                orig_input = fut_to_input.get(fut, "<unknown>")
                print(f"输入 {orig_input} 的任务已完成，结果: {res}")
            except Exception as e:
                # 对异常的处理策略可以是记录日志、告警、重试等。
                # 重要的是不要让未捕获的异常终止对其他已完成任务的处理循环。
                orig_input = fut_to_input.get(fut, "<unknown>")
                print(f"输入 {orig_input} 的任务失败：", type(e).__name__, e)

    # 线程池退出（with 语句结束）会等待已提交的任务完成或线程池清理完毕，
    # 因此到达这里时所有任务都已终结（成功或失败）。

def timeout_example():
    """演示 future.result(timeout=...) 超时处理。

    说明（详细）：
    - exe.submit(...) 立即返回一个 concurrent.futures.Future 对象，表示已安排的异步任务。
    - fut.result(timeout=sec) 会阻塞等待结果，最多等待 timeout 秒；超时则抛出 concurrent.futures.TimeoutError。
    - fut.cancel() 只会在任务尚未开始执行时成功（返回 True）。如果任务已经开始运行或已完成，cancel() 返回 False。
    - 即便在 with ThreadPoolExecutor(...) 语句块结束时，也会调用 executor.shutdown(wait=True) 等待正在运行的任务结束（除非显式调用 shutdown(wait=False)）。
    """
    with ThreadPoolExecutor(max_workers=2) as exe:
        # 提交一个故意较慢的任务：worker 会在 1.0-1.5 秒之间睡眠然后返回结果
        fut = exe.submit(worker, 100, (1.0, 1.5))

        try:
            # 以很短的超时尝试获取结果，这里很可能触发 TimeoutError
            print("尝试获取结果（短超时，0.1s）...")
            print("result:", fut.result(timeout=0.1))  # 若在 0.1s 内没有结果会抛出 TimeoutError
        except TimeoutError:
            # 超时后的可选策略示例：
            #  - 取消任务（如果尚未开始）；
            #  - 记录超时并稍后重试或继续等待（再次调用 result 或不传 timeout）。
            print("获取结果超时，可选择取消任务或继续等待。")
            # 尝试取消任务。注意：如果任务已经在某个工作线程中开始执行，cancel() 会返回 False。
            cancelled = fut.cancel()
            print("cancelled:", cancelled)
            # 如果取消失败（任务已在运行），可以选择在此阻塞等待其完成：
            # try:
            #     print("继续等待直到完成：", fut.result())
            # except Exception as e:
            #     print("任务在等待时抛出异常：", e)
        except Exception as e:
            # 捕获任务执行过程中抛出的异常（worker 内部异常会在这里重新抛出）
            print("任务异常：", e)

        # 注意：当退出 with 块时，ThreadPoolExecutor 会等待所有正在运行的任务结束。
        # 如果上面 cancel() 返回 False 且我们没有再次等待，退出时仍会阻塞直到该任务完成。

def shared_state_example():
    """演示多个线程修改共享可变状态时使用 Lock 保护。
    
    重点说明（简要）：
    - 当多个线程需要读/改/写同一个可变对象时（例如计数器），必须使用同步原语（如 Lock）
      来避免竞态条件（race condition）。即使 Python 有 GIL，复合操作（读 -> 修改 -> 写）
      也不是原子的，仍然会发生数据竞争。
    - 这里使用一个 dict 来保存计数器是为了让内部函数能够通过闭包修改它（字典是可变的）。
      也可以使用 list、类封装或使用全局变量来达到类似效果。
    - 在 incr 内部我们在临界区（with lock）里执行最小量的操作：读当前值、更新值。
      将尽量避免在持锁期间做耗时或阻塞操作，以减少对并发性能的影响。
    - time.sleep(0.001) 用来放大没有加锁时可能出现的竞态，从而使示例更易观察到问题。
    """
    # 共享的可变状态——计数器，使用字典以便在闭包中可修改
    counter = {"value": 0}
    # 用于保护对共享状态的访问
    lock = threading.Lock()

    def incr(n):
        """对 counter['value'] 执行 n 次自增操作，使用锁保护每次读/改/写。"""
        for _ in range(n):
            # 获取锁，进入临界区。with 语句确保即使发生异常也会释放锁。
            with lock:
                # 读取当前值（此读操作和后续写操作必须在锁内完成以保证一致性）
                tmp = counter["value"]
                # 模拟更复杂/耗时的中间步骤，放在锁外通常能提升并发性；
                # 这里故意放在锁内以演示如果在临界区做耗时工作会降低并发效率。
                time.sleep(0.001)
                # 写回更新后的值
                counter["value"] = tmp + 1

    # 每个任务增加的次数，三个任务总计应增加 3000 次
    increments = [1000, 1000, 1000]

    # 创建线程池并并发执行 incr。注意：
    # - executor.map 会立即提交所有任务并返回一个结果迭代器。
    # - 为了显式等待所有任务完成并忽略返回值（incr 返回 None），使用 list() 消耗迭代器。
    # - 也可以不调用 list()，仅依赖 with 语句退出时 executor.shutdown(wait=True) 来等待任务完成。
    with ThreadPoolExecutor(max_workers=3) as exe:
        list(exe.map(incr, increments))

    # 理论上应为 3000；如果未使用锁，常会得到小于 3000 的值（由于丢失更新）。
    print("最终计数（应为 3000）：", counter["value"])

def main():
    random.seed(0)
    print("示例：map")
    simple_map_example()
    print("\n示例：submit + callback")
    submit_with_callback_example()
    print("\n示例：as_completed")
    as_completed_example()
    print("\n示例：超时处理")
    timeout_example()
    print("\n示例：共享状态（加锁）")
    shared_state_example()
    print("\n线程池示例运行完毕。")

if __name__ == "__main__":
    main()